package com.tcm.kafka.thread;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.*;

public class FirstMultiConsumerThreadDemo {

    public static final String brokeList = "localhost:9092";
    public static final String topic = "topic-demo";
    public static final String groupId = "group.demo";

    public static Properties initConfig(){
        Properties properties = new Properties();
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("bootstrap.servers", brokeList);
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        // 设置消费组的名称
        properties.put("group.id", groupId);
        return properties;
    }

    public static void main(String[] args) {
        Properties properties = initConfig();
        int consumerThreadNum = 4;
        for(int i=0; i<consumerThreadNum; i++){
            new KafkaConsumerThread(properties, topic).start();
        }
    }

    static class KafkaConsumerThread extends Thread{

        private KafkaConsumer<String, String> consumer;

        public KafkaConsumerThread() {
        }

        public KafkaConsumerThread(Properties properties, String topic) {
            this.consumer = new KafkaConsumer<String, String>(properties);
            this.consumer.subscribe(Arrays.asList(topic));
        }

        @Override
        public void run() {
            try{
                while(true){
                    ConsumerRecords<String, String> records =
                            consumer.poll(Duration.ofMillis(100));
                    for(ConsumerRecord<String, String> record:records){
                        System.out.println(Thread.currentThread().getName()+"-------->"+record.value());
                    }
                }
            }catch (Exception e){
                e.printStackTrace();
            }finally {
                consumer.close();
            }
        }
    }

}
